package com.salmon.study.actors;


import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.UntypedAbstractActor;
import com.salmon.study.message.Result;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class MasterActor extends UntypedAbstractActor {

    private ActorRef aggregateActor = getContext().actorOf(Props.create(AggregateActor.class), "aggregateActor");

    private ActorRef reduceActor = getContext().actorOf(Props.create(ReduceActor.class, aggregateActor), "reduceActor");

    private ActorRef mapActor = getContext().actorOf(Props.create(MapActor.class, reduceActor), "mapActor");

    @Override
    public void onReceive(Object message) throws Throwable {
        log.info("{}", message);
        if(message instanceof String) {
            String data = (String) message;
            mapActor.tell(data, ActorRef.noSender());
        } else if (message instanceof Result) {
            aggregateActor.tell(message, ActorRef.noSender());
        } else {
            unhandled(message);
        }
    }
}
